package org.fastsyncer.manager.framework;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.fastsyncer.common.DefaultThreadFactory;
import org.fastsyncer.common.constant.CommonConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.exception.DataCenterException;
import org.fastsyncer.common.exception.TaskDistributorException;
import org.fastsyncer.common.util.ApplicationUtil;
import org.fastsyncer.manager.log.LogFactory;
import org.fastsyncer.manager.splitter.SplitterFactory;
import org.fastsyncer.manager.util.CleanUtil;
import org.fastsyncer.worker.framework.WorkerFactory;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ManagerFactory:
 * <p>执行流程如下：1.分析任务配置 2.持久化任务信息 3.计算拆分任务 4.指派任务给worker 5.接受worker执行结果 6.刷新最新进度</p>
 * 
 * @author AE86
 * @date 2017年5月13日 下午8:06:44
 * @version 1.0.0
 */
public final class ManagerFactory {

    private static final Logger logger = LoggerFactory.getLogger(ManagerFactory.class);

    private static volatile ManagerFactory instance = null;

    private ManagerFactory() {
    }

    public static ManagerFactory getInstance() {
        if (null == instance) {
            synchronized (ManagerFactory.class) {
                if (null == instance) {
                    instance = new ManagerFactory();
                }
            }
        }
        return instance;
    }

    // 支持同时处理数据同步任务并发数
    private static final Integer maxSync = Integer.parseInt(ApplicationUtil.getKey(CommonConstant.MANAGER_SYNC_NUMBER));

    // 支持同时处理数据迁移任务并发数
    private static final Integer maxMigrate = Integer.parseInt(ApplicationUtil.getKey(CommonConstant.MANAGER_MIGRATE_NUMBER));

    // 待执行的数据同步任务
    private Queue<JSONObject> syncQueue = new LinkedBlockingQueue<JSONObject>(maxSync);

    // 待执行的数据迁移任务
    private Queue<String> migrateQueue = new ConcurrentLinkedQueue<String>();

    // 正在执行的数据迁移任务
    private Map<String, String> running = new ConcurrentHashMap<String, String>();

    // 定义任务处理线程池,当第执行二个任务时,第一个任务完成,会复用线程,不会新建线程,执行结束一段时间会自动回收
    private ExecutorService pool = Executors.newCachedThreadPool(new DefaultThreadFactory("Manager-threadpool"));

    // 当前正在运行同步任务数
    private volatile Integer run = 0;
    
    private DataFactory data = DataFactory.getInstance();
    
    private LogFactory log = LogFactory.getInstance();
    
    private WorkerFactory worker = WorkerFactory.getInstance();

    private SplitterFactory splitter = SplitterFactory.getInstance();
    
    // 数据迁移
    public void handle(MappingTask mappingTask) throws Exception {
        // 1.分析任务配置
        if (mappingTask == null) {
            logger.error("The task configuration can not be empty!");
            return;
        }
        logger.info("The task is accepted.");

        // 2.计算总进度
        splitter.countProcess(mappingTask);

        // 3.持久化任务
        boolean store = data.saveMapping(mappingTask.getId(), mappingTask);
        if (!store) {
            logger.error("The persistent task failed!");
            return;
        }
        logger.info("The persistence task is complete.");

        // 4.检查执行任务状态
        this.validateTask(mappingTask.getId());

        // 5.添加任务到队列中
        if (migrateQueue.isEmpty()) {
            migrateQueue.offer(mappingTask.getId());
            // 触发迁移线程线程
            pool.execute(new TouchMigrateHandleThread());
        } else {
            migrateQueue.offer(mappingTask.getId());
        }
        logger.info("Added task to the queue.\r");
    }

    // 数据同步
    public void handle(JSONObject t) throws Exception {
        // 这里是数据增量同步处理，这里会接收增删改操作的数据。
        // 其中，针对DB操作，我们可以接收到同步前和同步后的变化，
        // 暂时直接同步。(这里会在1.0.0版本之后实现审计功能。)
        String taskId = t.getString("taskId");
        // 1.获取对应的映射
        MappingTask mappingTask = data.getMapping(taskId);
        if (null == mappingTask) {
            throw new Exception("驱动配置:" + taskId + "未找到.");
        }
        // 2.检查驱动是否被禁用
        if (!mappingTask.isEnable()) {
            throw new Exception("驱动配置:" + taskId + "已禁用.");
        }
        // 3.检查驱动是否为数据同步模式
        if (!mappingTask.getModel().equals(MappingConstant.SYNC_INCREMENT)) {
            throw new Exception("驱动配置必须为数据同步模式.");
        }
        // 4.过滤数据,如果没有一条满足同步条件,直接拒绝
        CleanUtil.clean(mappingTask, t);
        if (t.getJSONArray("msg").length() < 1) {
            return;
        }
        // 5.触发调度
        if (syncQueue.isEmpty()) {
            // 添加数据同步任务
            addTask(t);
            // 触发同步线程线程
            pool.execute(new TouchSyncHandleThread());
        } else {
            // 添加数据同步任务
            addTask(t);
        }
    }
    
    /**
     * 检查执行任务状态
     * @Title: validateTask 
     * @Description: 检查执行任务状态
     * @param driverId
     * @throws Exception
     * @return: void
     */
    public void validateTask(String driverId) throws IllegalArgumentException {
        // 检查是否已在执行
        if (null != running.get(driverId)) {
            throw new IllegalArgumentException("正在运行该任务！");
        }
        // 检查是否已添加该执行任务
        if (migrateQueue.contains(driverId)) {
            throw new IllegalArgumentException("已启动该任务！");
        }
        // 检查系统运行任务数是否堵塞
        if (run + 1 > maxMigrate) {
            throw new IllegalArgumentException("系统繁忙,请稍后再试!已运行任务:" + run);
        }
    }

    /**
     * 添加执行数据同步任务
     * 
     * @throws Exception
     */
    private void addTask(JSONObject t) throws Exception {
        // 排号检查是否添加成功，如果失败，直接抛出异常
        if (!syncQueue.offer(t)) {
            throw new Exception("系统繁忙,并发上限为" + maxSync + ",请稍后再试！");
        }
    }

    /**
     * 强占同步队列
     */
    private synchronized void occupySyncTask() throws Exception {
        if (run + 1 > maxMigrate) {
            throw new Exception("The system is busy, please try again later! Is currently running:" + run);
        }
        run += 1;
        logger.info("Occupy the queue >> available:" + (maxMigrate - run));
    }

    /**
     * 释放强占队列
     */
    private synchronized void releaseSyncTask() {
        run -= 1;
        logger.info("Release the queue >> available:" + (maxMigrate - run));
    }

    /**
     * 获取正在排队的同步任务
     * 
     * @return
     */
    public Queue<String> getQueue() {
        return migrateQueue;
    }

    /**
     * 获取正在运行的同步任务
     * 
     * @return
     */
    public Map<String, String> getRunning() {
        return running;
    }

    /**
     * 获取正在运行的同步任务
     * 
     * @return
     */
    public String getRunning(String taskId) {
        return running.get(taskId);
    }

    /**
     * 移除正在运行的同步任务
     * 
     * @return
     */
    public boolean removeRunning(String taskId) {
        running.remove(taskId);
        return true;
    }

    /**
     * 移除正在排队的同步任务
     * 
     * @return
     */
    public boolean removeQueue(String taskId) {
        migrateQueue.remove(taskId);
        return true;
    }

    /**
     * 触发数据迁移线程
     */
    class TouchMigrateHandleThread extends Thread {

        @Override
        public void run() {
            String taskId;
            while (!migrateQueue.isEmpty()) {
                try {
                    // 强占同步队列
                    occupySyncTask();
                    // 线程处理同步任务
                    taskId = migrateQueue.poll();
                    // 添加运行任务
                    running.put(taskId, taskId);
                    this.sync(taskId);
                    logger.info("The task has been finished successfully!!==============================>>" + taskId);

                    // 移除运行任务
                    running.remove(taskId);
                    // 释放强占队列
                    releaseSyncTask();
                } catch (Exception e) {
                    logger.error("An exception occurred while executing the task:" + e.getLocalizedMessage());
                }
            }
        }

        private void sync(String taskId) {
            try {
                MappingTask task = data.getMapping(taskId);
                // 总进度
                Integer process = task.getProcess();
                // 当前已完成进度
                Integer processCompleted = task.getProcessCompleted();
                long begin = System.currentTimeMillis();
                // 计算运行次数
                int runProcess = process - processCompleted;
                for (int k = 0; k < runProcess; k++) {
                    // ########检测任务已被删除时，中止任务########
                    if (null == getRunning(task.getId())) {
                        throw new Exception("The current task " + task.getId() + " has been terminated!");
                    }
                    // 2.数据迁移，将数据源的数据同步至目标源
                    execute(task);
                }
                // 更新完成时间
                MappingTask mapping = data.getMapping(taskId);
                mapping.setRunTime(System.currentTimeMillis() - begin);
                data.saveMapping(taskId, mapping);
            } catch (Exception e) {
                logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
            }
        }
        
        private void execute(MappingTask mappingTask) throws Exception {
            logger.info("Execute...");
            // 1.计算拆分任务，返回待执行的任务范围[100000000,100000999],[100001000,100001999],...
            Queue<Integer[]> distributeTask = splitter.split(mappingTask);
            if (null == distributeTask || distributeTask.isEmpty()) {
                throw new TaskDistributorException("Split task failed!");
            }
            
            // 目标源驱动
            String id = mappingTask.getId();
            int success = 0;
            int fail = 0;
            try {
                JSONObject handle = worker.handle(mappingTask, distributeTask);
                // 如果有错误
                fail = handle.getInt("fail");
                if (0 < fail) {
                    fail = mappingTask.getFail() + fail;
                    // 存储错误日志
                    log.handle(handle);
                }
                // 获取总数和成功数
                success = mappingTask.getSuccess() + handle.getInt("success");
            } catch (InterruptedException | JSONException e) {
                logger.error("An exception when performing data migration routing tasks:" + e.getMessage(), e.getCause());
            }
            
            // 3.更新完成任务进度
            mappingTask.setProcessCompleted(mappingTask.getProcessCompleted() + 1);
            mappingTask.setSuccess(success);
            mappingTask.setFail(fail);

            // 4.持久化任务信息
            boolean store = data.saveMapping(id, mappingTask);
            if (!store) {
                throw new DataCenterException("持久化任务存储失败!");
            }
            logger.info("Refresh task successfully!\r");
        }

    }

    /**
     * 触发数据同步线程
     */
    class TouchSyncHandleThread extends Thread {

        @Override
        public void run() {
            while (!syncQueue.isEmpty()) {
                try {
                    // 1.获取第一个元素
                    JSONObject t = syncQueue.poll();
                    
                    // 2.数据同步:将数据源的数据同步至目标源
                    sync(t, true);
                } catch (JSONException e) {
                    logger.error("An exception when routing task：" + e.getLocalizedMessage());
                } catch (Exception e) {
                    logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
                }
            }
        }
        
    }
    
    /**
     * 同步消息
     * @param msg
     * @param writeLog 是否记录错误日志
     * @return
     * @throws JSONException
     */
    private JSONObject sync(JSONObject msg, boolean writeLog) throws JSONException{
        if (null != msg) {
            String id = msg.getString("taskId");
            MappingTask t = data.getMapping(id);
            JSONObject handle = worker.handle(t, msg.getJSONArray("msg"));
            // 存在错误日志
            int fail = handle.getInt("fail");
            if (0 < fail) {
                t.setFail(t.getFail() + fail);
                if(writeLog){
                    log.handle(handle);
                }
            }
            int success = t.getSuccess() + handle.getInt("success");
            t.setSuccess(success);
            data.saveMapping(id, t);
            return handle;
        }
        return null;
    }
    
    /**
     * 如果同步成功返回Null
     * @param msg
     * @return
     * @throws JSONException
     */
    public JSONArray send(JSONObject msg) throws JSONException{
        JSONObject handle = sync(msg, false);
        int fail = handle.getInt("fail");
        if(0 < fail){
            return handle.getJSONArray("reason");
        }
        return null;
    }
}
